package com.qin.service;

import com.google.common.collect.Maps;
import com.qin.config.LogConfig;
import com.qin.handler.MyClientHandler;
import com.qin.properties.LogProperties;
import com.qin.protobuf.LogClientSendRespProto;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;

/**
 * 业务实现类
 *
 * @author qinxh
 * @date 2020/08/12
 */
public class LogService {

    private Channel channel;

    private String nettyServerInfo;

    private NioEventLoopGroup eventExecutors;

    private CuratorFramework client = LogConfig.getClient();

    private LogProperties logProperties = LogConfig.getLogProperties();

    public LogService() throws Exception {

        // 启动netty客户端
        startNettyClient();
        Thread.sleep(1000);
        // 监听zk指定节点的子节点
        permanentChildrenNodesWatch(logProperties.getZkNodePath());
    }

    /**
     * 启动netty客户端（多线程异步启动）
     */
    private synchronized void startNettyClient() {

        if(eventExecutors != null && !eventExecutors.isShutdown()){
            return;
        }

        CompletableFuture.runAsync(() -> {
            eventExecutors = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap
                        .group(eventExecutors)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch){
                                ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                                ch.pipeline().addLast(
                                        new ProtobufDecoder(LogClientSendRespProto.LogClientSendResp.getDefaultInstance()));
                                ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                                ch.pipeline().addLast(new ProtobufEncoder());
                                // 添加handler
                                ch.pipeline().addLast(new MyClientHandler());
                            }
                        });
                // 获取负载均衡策略后得到的netty服务端地址信息
                nettyServerInfo = getNettyServerInfo();
                String[] split = nettyServerInfo.split(":");
                String ip = split[0];
                int port = Integer.parseInt(split[1]);
                ChannelFuture channelFuture = bootstrap.connect(ip, port).sync();
                System.out.println("netty客户端成功注册到：" + nettyServerInfo + "服务节点");
                channel = channelFuture.channel();
                channelFuture.channel().closeFuture().sync();
            } catch (Exception e) {
                System.out.println("netty客户端连接异常：" + e.getMessage());
            } finally {
                eventExecutors.shutdownGracefully();
            }
        });
    }

    /**
     * 获取 netty 客户端通道
     *
     * @return Channel
     */
    public Channel getChannel() {

        return channel;
    }

    /**
     * 设置负载均衡策略（服务端连接客户端个数 + 服务端当天日志文件大小 + 权重值 三个维度决定提供连接的服务节点）
     */
    private String getNettyServerInfo() throws Exception {

        List<String> childrenNodes = getChildrenNodes();
        if(CollectionUtils.isEmpty(childrenNodes)){
            throw new RuntimeException("无可用netty服务");
        }
        // 1. 获取日志文件和客户端连接数权重值（范围为1 ~ 9，默认值5。用于计算netty连接负载策略计算。如果超过此范围，使用默认值5）
        int sizeAndNumWeights = 5;
        if(logProperties.getClientNumber() != null && logProperties.getSizeAndNumWeights() > 0 && logProperties.getSizeAndNumWeights() < 10){
            sizeAndNumWeights = logProperties.getSizeAndNumWeights();
        }
        // 2. 根据 值_节点名 得到Map关系
        Map<Integer, String> clientNumberStrMap = Maps.newHashMap();
        Map<Integer, String> dataSizeStrMap = Maps.newHashMap();
        String regex = ",";
        for (String childNodeName : childrenNodes) {
            if(!existNode(childNodeName)){
                throw new RuntimeException("节点[" + childNodeName + "]信息不存在");
            }
            DistributedAtomicInteger integer = new DistributedAtomicInteger(
                    client,logProperties.getClientNumber() + childNodeName, LogConfig.getRetryPolicy());

            clientNumberStrMap.merge(integer.get().postValue(),childNodeName + regex, (x, y) -> x + y);
            dataSizeStrMap.merge(Integer.valueOf(getNode(logProperties.getZkNodePath() + "/" + childNodeName))
                    ,childNodeName + regex, (x, y) -> x + y);
        }
        // 3. 根据权重值得到每个服务节点计算权重后的值
        //map转换成list进行排序
        List<Map.Entry<Integer, String>> numberStrList = new ArrayList<>(clientNumberStrMap.entrySet());
        List<Map.Entry<Integer, String>> sizeStrList = new ArrayList<>(dataSizeStrMap.entrySet());
        //根据value值排序
        numberStrList.sort(Comparator.comparing(Map.Entry::getKey));
        sizeStrList.sort(Comparator.comparing(Map.Entry::getKey));
        //获取每个节点计算权重后的值
        Map<String, Integer> weightsMap = Maps.newHashMap();
        for (int i = 0; i < numberStrList.size(); i++) {
            String[] split = numberStrList.get(i).getValue().split(regex);
            for (String s : split) {
                weightsMap.put(s, (i + 1) * sizeAndNumWeights);
            }
        }
        for (int i = 0; i < sizeStrList.size(); i++) {
            String[] split = sizeStrList.get(i).getValue().split(regex);
            for (String s : split) {
                weightsMap.merge(s,(i + 1) * (10 - sizeAndNumWeights), Integer::sum);
            }
        }
        // 4. 得到值最小的节点即为当前所需服务节点信息
        List<Map.Entry<String, Integer>> entryArrayList = new ArrayList<>(weightsMap.entrySet());
        entryArrayList.sort(Comparator.comparing(Map.Entry::getValue));

        return Optional.ofNullable(entryArrayList.get(0).getKey()).orElseThrow(() -> new RuntimeException("获取netty服务端地址信息失败"));
    }

    /**
     * 获取指定netty日志收集服务端挂载zk的节点路径的子节点列表
     *
     * @return List<String>     子节点列表
     */
    private List<String> getChildrenNodes() throws Exception {

        return client.getChildren().forPath(logProperties.getZkNodePath());
    }

    /**
     * 获取节点信息
     *
     * @param path  节点名称
     */
    private String getNode(String path) throws Exception {

        Stat stat = new Stat();
        byte[] data = client.getData().storingStatIn(stat).forPath(path);

        return new String(data);
    }

    /**
     * 判断节点是否存在
     *
     * @param childNodeName  节点名称
     */
    private Boolean existNode(String childNodeName) throws Exception {

        // 如果节点存在则返回其状态信息如果不存在则为 null
        Stat stat = client.checkExists().forPath(logProperties.getZkNodePath() + "/" + childNodeName);

        return Optional.ofNullable(stat).isPresent();
    }

    /**
     * 监听zk指定节点的子节点
     *
     */
    private void permanentChildrenNodesWatch(String nodePath) throws Exception {

        // 第三个参数代表除了节点状态外，是否还缓存节点内容
        PathChildrenCache childrenCache = new PathChildrenCache(client, nodePath, true);
        /*
         * 创建完pathChildrenCache,一定记得调用start方法!!!不然是不会生效的
         *
         * StartMode 代表初始化方式:
         *    NORMAL: 异步初始化；
         *    BUILD_INITIAL_CACHE: 同步初始化；
         *    POST_INITIALIZED_EVENT: 异步并通知，初始化之后会触发 INITIALIZED 事件
         */
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        // 创建PathChildrenCache如果比较多的话,一定要记得自定义其使用的线程池参数,不然每次new出来一个PathChildrenCache,
        // 就会自行创建一个单线程池,创建不了多少就会开始抛超最多线程的个数异常。这时使用 addListener(T listener, Executor executor) 方法
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
                switch (event.getType()) {
                    case INITIALIZED:
                        System.out.println("监听[ "+ nodePath + " ]节点的子节点，初始化完成");
                        break;
                    // 新增
                    case CHILD_ADDED:
                        // TIPS：增加netty服务节点后，且netty客户端没有启动，那么netty客户端需要重新连接netty服务节点
                        if(eventExecutors == null || eventExecutors.isShutdown()){
                            System.out.println("增加服务节点:" + event.getData().getPath() + "，且触发本netty客户端启动!");
                            startNettyClient();
                        }
                        break;
                    // 删除
                    case CHILD_REMOVED:
                        String path = event.getData().getPath();
                        // TIPS：如果是本客户端连接的节点被删除，那么需要切换netty服务节点
                        String[] split = path.split("/");
                        if(split[split.length - 1].equals(nettyServerInfo) ){
                            System.out.println("服务节点:" + path + "退出，且为本netty客户端的服务节点。准备尝试切换其他节点!");
                            // 关闭本次netty客户端
                            if(!eventExecutors.isShutdown()){
                                eventExecutors.shutdownGracefully();
                            }
                            // 关闭通道
                            Channel channel1 = getChannel();
                            if(channel1 != null && channel1.isOpen()){
                                channel1.close();
                            }
                            // 将被删除的netty服务节点信息置空
                            nettyServerInfo = "";
                            // 重新启动客户端
                            startNettyClient();
                        }
                        break;
                    // 修改
                    case CHILD_UPDATED:
                        ChildData data = event.getData();
                        System.out.println("服务节点[ " + data.getPath() + " ]修改后数据:" + new String(data.getData()));
                        break;
                    default:
                }
            }
        });
    }
}
